package com.atguigu.windows;

import com.atguigu.bean.WaterSensor;
import com.atguigu.datasource.RandomWatersensor;
import com.atguigu.utils.ITtoListUtil;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Date;
import java.util.List;

public class Flink09_WaterMarkCustom {
    public static void main(String[] args) {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 10000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);
        env.setParallelism(1);

        //设置水印发送状态为周期性时的间隔时间
        env.getConfig().setAutoWatermarkInterval(3000);
        env.addSource(new RandomWatersensor())
                //加水印
                .assignTimestampsAndWatermarks(WatermarkStrategy

                        //此处调用forGenerator生成自定义水印生成器
                        .forGenerator(new WatermarkStrategy<WaterSensor>() {
                            @Override
                            public WatermarkGenerator<WaterSensor> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
                                //此处需要返回一个WatermarkGenerator
                                // 那么我们返回自定义的水印生成器
                                return new MyWaterMarkGenerator();

                            }
                        })
                        //指定那个字段作为水印比较时间戳
                        .withTimestampAssigner(TimestampAssignerSupplier.of(new SerializableTimestampAssigner<WaterSensor>() {
                            @Override
                            public long extractTimestamp(WaterSensor element, long recordTimestamp) {
                                //返回需要进行比较的水印比较时间戳
                                return element.getTs();
                            }
                        }))

                )
                //进行keyBy操作
                .keyBy(new KeySelector<WaterSensor, String>() {
                    @Override
                    public String getKey(WaterSensor value) throws Exception {
                        return value.getId();
                    }
                })
                //进行窗口处理
                //这里是基于事件时间的开窗
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                //对数据进行处理
                .process(new ProcessWindowFunction<WaterSensor, String, String, TimeWindow>() {
                    @Override
                    public void process(String key,
                                        Context context,
                                        Iterable<WaterSensor> elements,
                                        Collector<String> out) throws Exception {
                        String start = new Date(context.window().getStart()).toLocaleString();
                        String end = new Date(context.window().getEnd()).toLocaleString();
                        System.out.println("窗口起始：" + start + ",窗口结束：" + end);
                        //输出元素
                        List<WaterSensor> list = ITtoListUtil.toList(elements);

                        //元素原样输出
                        out.collect(list.toString());
                    }
                })
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


    //自定义水印生成器
    public static class MyWaterMarkGenerator implements WatermarkGenerator<WaterSensor> {

        //定义一个初始最大值水印,这个3000是最大容忍迟到时间
        long maxTs = Long.MIN_VALUE + 3000;

        //这个是间歇性水印生成器
        //来一条数据就会执行一次该方法
        @Override
        public void onEvent(WaterSensor event, long eventTimestamp, WatermarkOutput output) {
            //我们在这里会处理每一条来的数据
            //获取数据中的时间戳，并且重新定于最大值水印
            maxTs = Math.max(maxTs, event.getTs());
            //如果我们想要生成间歇性的水印生成器，那么就在这里进行水印输出，就不需要在onPeriodicEmit进行水印输出了
            output.emitWatermark(new Watermark(maxTs - 3000));

        }

        //这个是周期性水印生成器
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            //默认每隔200毫秒就发送一次水印
            //如果想要自定义水印发送的时间间隔，在env配置中设置
            //env.getConfig().setAutoWatermarkInterval(3000);

            System.out.println("MyWaterMarkGenerator.onPeriodicEmit");
        }
    }
}
